package cn.doitedu.dashboard;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job02_PageTypeTopUV_page {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///D:/flink/ck");

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 建表，映射 dwd层的用户行为明细宽表
        tenv.executeSql(
                         "create table dwd_events_kafka(     " +
                        "     event_id           string             " +
                        "    ,action_time        bigint             " +
                        "    ,user_id            bigint             " +
                        "    ,page_type          string             " +
                        "    ,properties        map<string,string>  " +
                        "    ,url as properties['url']              " +
                        "    ,rt as to_timestamp_ltz(action_time,3)   " +
                        "    ,watermark for rt as rt - interval '0' second   " +
                        ") WITH (                      " +
                        "    'connector' = 'kafka',    " +
                        "    'topic' = 'dwd-user-action-log',  " +
                        "    'properties.bootstrap.servers' = 'doitedu:9092',  " +
                        "    'properties.group.id' = 'doit44_g1',              " +
                        "    'scan.startup.mode' = 'latest-offset',            " +
                        "    'value.format' = 'json',                          " +
                        "    'value.fields-include' = 'EXCEPT_KEY'             " +
                        ")                                                     ");


        tenv.executeSql(
                " create table sink_mysql(                                 "+
                        "     page_type  string,                        "+
                        "     rn  bigint,                            "+
                        "     window_start timestamp(3),                     "+
                        "     window_end timestamp(3),                     "+
                        "     url string,                                   "+
                        "     uv bigint,                                   "+
                        "     primary key(page_type,rn) not enforced         "+
                        " ) with (                                          "+
                        "  'connector' = 'jdbc',                            "+
                        "  'url' = 'jdbc:mysql://doitedu:3306/doit47',      "+
                        "  'table-name' = 'dashboard_pagetype_topn',        "+
                        "  'username' = 'root',                             "+
                        "  'password' = 'root'                              "+
                        " )                                                 "
        );




        tenv.executeSql(
                "insert into sink_mysql " +
                "with tmp1 as (\n" +
                "    select\n" +
                "        window_start,\n" +
                "        window_end,\n" +
                "        page_type,\n" +
                "        url,\n" +
                "        count(distinct user_id) as uv\n" +
                "    from table(\n" +
                "        tumble(table dwd_events_kafka , descriptor(rt),interval '5' minute)\n" +
                "    )\n" +
                "    group by \n" +
                "        window_start,\n" +
                "        window_end,\n" +
                "        page_type,\n" +
                "        url\n" +
                ")    \n" +
                ",tmp2 as (\n" +
                "select\n" +
                "    window_start,\n" +
                "    window_end,\n" +
                "    page_type,\n" +
                "    url,\n" +
                "    uv,\n" +
                "    row_number() over(partition by window_start,window_end,page_type order by uv desc ) as rn \n" +
                "from tmp1\n" +
                ") \n" +
                "select\n" +
                "    page_type,\n" +
                "    rn,       \n" +
                "    window_start,\n" +
                "    window_end,  \n" +
                "    url,  \n" +
                "    uv    \n" +
                "from tmp2 \n" +
                "where rn<=10");



    }
}
